package cn.me.alphamq.common.remote;

import cn.me.alphamq.common.model.client.TxCallbackCheckEvent;
import cn.me.alphamq.common.entity.BaseMsg;
import cn.me.alphamq.common.entity.req.TxCallbackCheckReq;
import cn.me.alphamq.common.entity.resp.*;
import cn.me.alphamq.common.enumeration.EventCodeEnum;
import cn.me.alphamq.common.event.EventBus;
import cn.me.alphamq.common.event.model.Event;
import cn.me.alphamq.common.msg.TcpMsg;
import com.alibaba.fastjson2.JSON;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;

/**
 * broker 响应的处理器，可共享
 * <p>
 * 除了响应之外，该处理器还可以处理 broker 的回调
 *
 * @author f
 */
@ChannelHandler.Sharable
public class BrokerRespHandler extends SimpleChannelInboundHandler<TcpMsg> {

    /**
     * 事件总线
     */
    private final EventBus eventBus;

    public BrokerRespHandler(EventBus eventBus) {
        this.eventBus = eventBus;
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TcpMsg tcpMsg) throws Exception {
        int code = tcpMsg.getCode();

        // 处理 broker 的响应
        Class<? extends BaseMsg> clazz = null;
        if (EventCodeEnum.PUSH_MSG_RESP.getCode() == code) {
            clazz = PushMsgResp.class;
        } else if (EventCodeEnum.CONSUME_MSG_RESP.getCode() == code) {
            clazz = ConsumeMsgResp.class;
        } else if (EventCodeEnum.CONSUME_SUCCESS_RESP.getCode() == code) {
            clazz = ConsumeSuccessResp.class;
        } else if (EventCodeEnum.CREATE_TOPIC_RESP.getCode() == code) {
            clazz = CreateTopicResp.class;
        } else if (EventCodeEnum.RETRY_CONSUME_MSG_RESP.getCode() == code) {
            clazz = RetryConsumeMsgResp.class;
        } else if (EventCodeEnum.PUSH_DELAY_MSG_RESP.getCode() == code) {
            clazz = PushDelayMsgResp.class;
        } else if (EventCodeEnum.PUSH_TX_MSG_RESP.getCode() == code) {
            clazz = PushTxMsgResp.class;
        }
        if (clazz != null) {
            SyncFuture.removeAndSetResp(clazz, tcpMsg);
            return;
        }

        // 处理 broker 的回调
        byte[] body = tcpMsg.getBody();
        Event event;
        if (EventCodeEnum.TX_CALLBACK_CHECK_REQ.getCode() == code) {
            event = new TxCallbackCheckEvent(JSON.parseObject(body, TxCallbackCheckReq.class));
        } else {
            throw new RuntimeException("invalid msg code: " + code);
        }
        event.setCtx(ctx);
        eventBus.publish(event);
    }
}
